package com.sopaths.crawler.thread;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import javax.inject.Named;

import org.apache.log4j.Logger;

import com.sopaths.contract.crawler.CrawlerTaskType;
import com.sopaths.crawler.task.handler.CrawlerTaskResultHandler;
import com.sopaths.crawler.tasks.CrawlerTask;

@Named
public class CrawlerTaskResultHandlerDispatcher {

	private static Logger logger = Logger
			.getLogger(CrawlerTaskResultHandlerDispatcher.class);

	private BlockingQueue<CrawlerTask<?>> queue;

	private Map<CrawlerTaskType, CrawlerTaskResultHandler> handlerMap;

	private int resultHandlerThreadCount;

	private String threadName;

	private int queueSize;

	public void Handle(CrawlerTask<?> task) {
		try {
			String serial = null;
			if (logger.isInfoEnabled()) {
				serial = UUID.randomUUID().toString();
				logger.info("Submit task result: " + task.getClass()
						+ ". UUID: " + serial);
			}
			queue.put(task);
			if (logger.isDebugEnabled()) {
				logger.debug("Submit task result successfully. Queue Size: "
						+ queue.size() + ". UUID: " + serial);
			}
		} catch (InterruptedException ex) {
			logger.error("An error occurred while submit result", ex);
		}
	}

	public void prepare() {
		queue = new LinkedBlockingQueue<>(queueSize);
		for (int i = 0; i < resultHandlerThreadCount; i++) {
			Thread thread = new Thread(
					new CrawlerTaskResultHandlerDispatcherThread(handlerMap));
			thread.setName(threadName + "-" + i);
			thread.start();
		}
	}

	private class CrawlerTaskResultHandlerDispatcherThread implements Runnable {

		private Logger logger = Logger
				.getLogger(CrawlerTaskResultHandlerDispatcherThread.class);

		private Map<CrawlerTaskType, CrawlerTaskResultHandler> handlerMap;

		public CrawlerTaskResultHandlerDispatcherThread(
				Map<CrawlerTaskType, CrawlerTaskResultHandler> handlerMap) {
			this.handlerMap = handlerMap;
		}

		@Override
		public void run() {
			while (true) {
				try {
					logger.info("Take task result from queue.");
					CrawlerTask<?> task = queue.take();
					if (logger.isDebugEnabled()) {
						logger.debug("Take task result from queue successfully. Task: "
								+ task.getClass()
								+ ". Queue size: "
								+ queue.size());
					}
					CrawlerTaskResultHandler resultHandler = handlerMap
							.get(task.getType());
					resultHandler.handle(task);
				} catch (Exception ex) {
					logger.error("An error occurred while handling result.", ex);
				}
			}
		}
	}

	public void setResultHandlerThreadCount(int resultHandlerThreadCount) {
		this.resultHandlerThreadCount = resultHandlerThreadCount;
	}

	public void setThreadName(String threadName) {
		this.threadName = threadName;
	}

	public void setQueueSize(int queueSize) {
		this.queueSize = queueSize;
	}

	public void setHandlerMap(
			Map<CrawlerTaskType, CrawlerTaskResultHandler> handlerMap) {
		this.handlerMap = handlerMap;
	}

}
